package spark

import java.util

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.util.{AccumulatorV2}

object MyAccumulator {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName(SimpleAcc.getClass.getSimpleName)
    val sc = new SparkContext(conf)
    val dataRDD: RDD[String] = sc.makeRDD(List("hadoop", "hbase", "hive", "scala", "spark"), 2)
    //todo 创建累加器
    val wordAccumulator = new WordAccumulator
    //todo 注册累加器
    sc.register(wordAccumulator)
    dataRDD.foreach {
      case word => {
        //todo 执行累加器功能
        wordAccumulator.add(word)
      }
    }
    //todo 获取累加器的值
    println("sum=" + wordAccumulator.value)
    //
  }
}


//声明累加器
//1.继承AccumulatorV2
//2.实现抽象方法
class WordAccumulator extends AccumulatorV2[String, util.ArrayList[String]] {
  val list = new util.ArrayList[String]()

  //当前累加器是否为初始化状态
  override def isZero: Boolean = list.isEmpty

  //复制累加器对象
  override def copy(): AccumulatorV2[String, util.ArrayList[String]] = {
    new WordAccumulator()
  }

  //重置累加器对象
  override def reset(): Unit = {
    list.clear()
  }

  //向累加器添加值
  override def add(v: String): Unit = {
    if (v.contains("h")) {
      list.add(v)
    }
  }

  //合并累加器
  override def merge(other: AccumulatorV2[String, util.ArrayList[String]]): Unit = {
    list.addAll(other.value)
  }

  //获取累加器的值
  override def value: util.ArrayList[String] = list
}